package com.tang.source;

import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.StringJoiner;

/**
 * FlinkCDCDemo
 *
 * @author tang
 * @since 2023/6/1 14:08
 */
public class MySqlSourceDemo {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        String databaseName = "lottery";
        int databaseCount = 2, databaseStart = 1;
        String tableName = "user_strategy_export";
        int tableCount = 4, tableStart;

        StringJoiner databaseJoin = new StringJoiner(",");
        StringJoiner tableJoin = new StringJoiner(",");
        for (int i = 0; i < databaseCount; i++) {
            String dataBaseName = databaseName + "_0" + databaseStart++;
            databaseJoin.add(dataBaseName);
            tableStart = 0;
            for (int i1 = 0; i1 < tableCount; i1++) {
                tableJoin.add(dataBaseName + "." + tableName + "_00" + tableStart++);
            }
        }
        System.out.println(databaseJoin);
        System.out.println(tableJoin);

        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("192.168.70.141")
                .port(3306)
                //.databaseList("lottery_01,lottery_02") // set captured database
                .databaseList(databaseJoin.toString()) // set captured database
                //.tableList("lottery_01.user_strategy_export_000,lottery_01.user_strategy_export_001,lottery_01.user_strategy_export_002,lottery_01.user_strategy_export_003,lottery_02.user_strategy_export_000,lottery_02.user_strategy_export_001,lottery_02.user_strategy_export_002,lottery_02.user_strategy_export_003") // set captured table
                .tableList(tableJoin.toString()) // set captured table
                .username("root")
                .password("123@abAB")
                .serverTimeZone("UTC") // 注意看报错，可能会提示时区不一致问题
                .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
                .build();

        // enable checkpoint
        env.enableCheckpointing(3000);

        env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source")
                // set 4 parallel source tasks
                .setParallelism(4)
                .print().setParallelism(1); // use parallelism 1 for sink to keep message ordering

        env.execute("Print MySQL Snapshot + Binlog");
    }

}
